package com.xujian.flink.source;

import com.xujian.common.beans.Stock;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

/**
 * @author star xu
 * @date 2023/1/27 18:24
 * @Description:
 * @Slogan: 致敬大师, 致敬未来的自己
 */
public class UdfSource {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        final DataStreamSource<Stock> orderDataStreamSource = env.addSource(new MyOrderSource());
        orderDataStreamSource.print();
        env.execute("UDFOrderSOurce");
    }

    public static class MyOrderSource implements SourceFunction<Stock> {
        private boolean running = true;

        @Override
        public void run(SourceContext<Stock> ctx) throws Exception {
            final Random random = new Random();
            while (running) {
                Stock stock = new Stock();
                stock.setId("stock_" + System.currentTimeMillis() % 700);
                stock.setPrice(random.nextDouble() * 100);
                stock.setStockName("UDFStock");
                stock.setTimestamp(System.currentTimeMillis());

                ctx.collect(stock);
                Thread.sleep(1000);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }
}
